/**
 * Copyright 2009 - Morten Udnæs
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.javabatch.petstore.process;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import com.javabatch.fileio.FieldError;
import com.javabatch.fileio.FileBuffer;
import com.javabatch.fileio.FileStore;
import com.javabatch.fileio.RecordBasedFileStore;
import com.javabatch.petstore.records.CatalogRecord;
import com.javabatch.petstore.records.CustomerRecord;
import com.javabatch.petstore.records.InventoryRecord;
import com.javabatch.petstore.records.InvoiceRecord;
import com.javabatch.petstore.records.OrderRecord;
import com.javabatch.petstore.records.Record;

/**
 * Animal record file containing id and animal name.
 * 
 * @author eco
 * 
 */
public class CreateInvoiceProcess implements BatchProcess {
	private FileStore orderStore;
	private FileStore customerStore;
	private FileStore catalogStore;
	private FileStore invoiceStore;

	private HashMap<String, Integer> catalogCache;
	

	public CreateInvoiceProcess(String orderFile, String customerFile, String catalogFile, String invoiceFile) {

		orderStore = new RecordBasedFileStore(orderFile, RecordBasedFileStore.READ_ACCESS, OrderRecord
				.getLengthAndValidateRecord());

		customerStore = new RecordBasedFileStore(customerFile, RecordBasedFileStore.READ_ACCESS, CustomerRecord
				.getLengthAndValidateRecord());

		catalogStore = new RecordBasedFileStore(catalogFile, RecordBasedFileStore.READ_ACCESS, CatalogRecord
				.getLengthAndValidateRecord());

		invoiceStore = new RecordBasedFileStore(invoiceFile, RecordBasedFileStore.WRITE_ACCESS, InventoryRecord
				.getLengthAndValidateRecord());
	}

	private void writeInvoiceRecord(Invoice theInvoice, List<FieldError> errors) {
		FileBuffer invoiceRecord = new FileBuffer();
		InvoiceRecord.INVOICE_ID.write(invoiceRecord, theInvoice.getInvoiceIdAsString());
		InvoiceRecord.NAME.write(invoiceRecord, theInvoice.getName());
		InvoiceRecord.ITEMS.write(invoiceRecord, theInvoice.getItemsAsString());
		InvoiceRecord.TOTAL.write(invoiceRecord, theInvoice.getTotalAsString()); 
		
		//Adding Linefeed for easy reading..
		InvoiceRecord.LINEFEED.write(invoiceRecord, "\n");

		invoiceStore.write(invoiceRecord);
	}

	private void openFiles() {
		orderStore.open();
		customerStore.open();
		invoiceStore.open();
	}

	private void closeFiles() {
		orderStore.close();
		customerStore.close();
		invoiceStore.close();
	}

	@Override
	public List<Object> prepare(List<Record> records) {
		List<FieldError> errors = new ArrayList<FieldError>();
		catalogCache = new HashMap<String, Integer>();
		catalogStore.open();

		FileBuffer catalogRecord = null;
		while (catalogStore.hasMore()) {
			catalogRecord = catalogStore.readNext();
			String catalogId = CatalogRecord.CATALOG_ID.read(catalogRecord, errors);
			String price = CatalogRecord.PRICE.read(catalogRecord, errors);
			catalogCache.put(catalogId, Integer.parseInt(price));
		}
		catalogStore.close();
		return null;
	}

	@Override
	public List execute(List<Record> records) {
		List<FieldError> errors = new ArrayList<FieldError>();
		Integer invoiceId = 1;
		long numOrders = 0;

		openFiles();
		
		FileBuffer orderRecord = getNextOrder();
		String customerIdOnOrder = OrderRecord.CUSTOMER_ID.read(orderRecord, errors);
		
		FileBuffer customerRecord = getNextCustomer(customerIdOnOrder, errors);
		String currentCustomerId = CustomerRecord.CUSTOMER_ID.read(customerRecord, errors);
		
		Invoice invoice = new Invoice(CustomerRecord.CUSTOMER_NAME.read(customerRecord, errors), invoiceId);

		while (orderRecord != null) {
					
			if (customerIdOnOrder.compareTo(currentCustomerId) > 0) {
				writeInvoiceRecord(invoice, errors);
				invoiceId++;
				customerRecord = getNextCustomer(customerIdOnOrder, errors);
				currentCustomerId = CustomerRecord.CUSTOMER_ID.read(customerRecord, errors);
				invoice = new Invoice(CustomerRecord.CUSTOMER_NAME.read(customerRecord, errors), invoiceId);
			}

			Integer price = catalogCache.get(OrderRecord.CATALOG_ID.read(orderRecord, errors));
			invoice.updateLine(price, OrderRecord.ITEMS.read(orderRecord, errors));
			
			orderRecord = getNextOrder();
			
			if (orderRecord != null) {
				customerIdOnOrder = OrderRecord.CUSTOMER_ID.read(orderRecord, errors);
				numOrders++;
			}
		} 
		
		if (invoice != null) {
			writeInvoiceRecord(invoice, errors);
		}

		if (errors.size() > 0) {
			throw new RuntimeException("File is missing customer data or is not sorted correctly!");
		}
		closeFiles();
		ArrayList<Long> result = new ArrayList<Long>();
		result.add(numOrders);
		result.add(new Long(invoiceId));
		return result;
	}

	private FileBuffer getNextCustomer(String customerIdOnOrder, List<FieldError> errors) {
		FileBuffer customerRecord = customerStore.readNext();
		
		String currentCustomerId = CustomerRecord.CUSTOMER_ID.read(customerRecord, errors);
		while (customerStore.hasMore() && currentCustomerId.compareTo(customerIdOnOrder) < 0) {
			customerRecord = customerStore.readNext();
			currentCustomerId = CustomerRecord.CUSTOMER_ID.read(customerRecord, errors);
		}

		if (!currentCustomerId.equals(customerIdOnOrder))
			throw new RuntimeException("Customer record is missing customerId: " + customerIdOnOrder);

		return customerRecord;
	}

	private FileBuffer getNextOrder() {
		FileBuffer orderRecord = orderStore.readNext();
		while (filter(orderRecord)) {
			orderRecord = orderStore.readNext();
		}
		return orderRecord;
	}

	private boolean filter(FileBuffer orderRecord) {
		return false;
	}

	@Override
	public void release(List<Record> records) {
		// TODO Auto-generated method stub

	}

}
